package chapter04

/**
 * author: yuhui
 * descriptions:
 * date: 2025 - 02 - 12 11:06 上午
 */
import org.apache.spark.{SparkConf, SparkContext}

object RDDCheckpointExample {
  def main(args: Array[String]): Unit = {
    // 初始化Spark配置和上下文
    val conf = new SparkConf()
      .setAppName("RDDCheckpointExample")
      .setMaster("local[*]")
    val sc = new SparkContext(conf)

    // 设置检查点目录
    sc.setCheckpointDir("hdfs:///path/to/checkpoint/dir")

    // 读取数据（例如，从一个文本文件中）
    val initialRDD = sc.textFile("hdfs:///path/to/initial/data.txt")

    // 执行一些转换操作
    val transformedRDD = initialRDD.map(line => {
      val parts = line.split(",")
      (parts(0), parts(1).toInt)
    })

    // 对转换后的RDD进行checkpoint
    transformedRDD.checkpoint()

    // 后续操作（例如，进行reduceByKey操作）
    val resultRDD = transformedRDD.reduceByKey(_ + _)

    // 第一次Action操作
    val result1 = resultRDD.collect()
    println("First result: " + result1.mkString(", "))

    // 后续可能还有更多的Action操作，它们都会重用checkpointed的RDD
    // ...

    // 停止SparkContext
    sc.stop()
  }
}